﻿using Apache.NMS;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;

namespace HAActiveMQ.Entity
{
    public class MqConsumer<TEntity> : BaseMq
        where TEntity : Interface.IMqReceiveMsg, new()
    {
        /// <summary>
        /// Listener方法
        /// </summary>
        public MethodInfo Method { get; set; }
        /// <summary>
        /// 消费者对象
        /// </summary>
        TEntity Entity { get; set; }
        /// <summary>
        /// 默认参数方便使用者进行测试
        /// </summary>
        /// <param name="ipAddress"></param>
        /// <param name="user"></param>
        /// <param name="password"></param>
        public MqConsumer(string ipAddress = "activemq:failover:(tcp://127.0.0.1:61616)", string user = "admin", string password = "admin")
        {
            Url = ipAddress;
            factory = new NMSConnectionFactory(Url);
            connection = factory.CreateConnection(user, password);
            connection.Start();
        }

        /// <summary>
        /// 添加Queue监听
        /// </summary>
        /// <param name="name"></param>
        /// <param name="method"></param>
        public void AddQueueListener(string name = "", MethodInfo method = null)
        {
            if (string.IsNullOrWhiteSpace(name)) name = this.Name;
            if (string.IsNullOrWhiteSpace(name)) return;
            if (method == null) method = this.Method;

            ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
            IDestination dest = session.GetQueue(name);
            IMessageConsumer consumer = session.CreateConsumer(dest);
            if (method != null)
            {
                consumer.Listener += new MessageListener(OnTextMessage);
            }
        }
        /// <summary>
        /// 添加Topic监听
        /// </summary>
        /// <param name="name"></param>
        /// <param name="method"></param>
        public void AddTopicListener(string name = "", MethodInfo method = null)
        {
            if (string.IsNullOrWhiteSpace(name)) name = this.Name;
            if (string.IsNullOrWhiteSpace(name)) return;
            if (method == null) method = this.Method;

            ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
            IDestination dest = session.GetTopic(name);
            IMessageConsumer consumer = session.CreateConsumer(dest);
            if (method != null)
            {
                consumer.Listener += new MessageListener(OnTextMessage);
            }
        }

        /// <summary>
        /// 添加Queue监听
        /// </summary>
        /// <param name="name"></param>
        /// <param name="method"></param>
        public void AddQueueListenerObj(string name = "", MethodInfo method = null)
        {
            if (string.IsNullOrWhiteSpace(name)) name = this.Name;
            if (string.IsNullOrWhiteSpace(name)) return;
            if (method == null) method = this.Method;

            ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
            IDestination dest = session.GetQueue(name);
            IMessageConsumer consumer = session.CreateConsumer(dest);
            if (method != null)
            {
                consumer.Listener += new MessageListener(OnObjectMessage);
            }
        }
        /// <summary>
        /// 添加Topic监听
        /// </summary>
        /// <param name="name"></param>
        /// <param name="method"></param>
        public void AddTopicListenerObj(string name = "", MethodInfo method = null)
        {
            if (string.IsNullOrWhiteSpace(name)) name = this.Name;
            if (string.IsNullOrWhiteSpace(name)) return;
            if (method == null) method = this.Method;

            ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
            IDestination dest = session.GetTopic(name);
            IMessageConsumer consumer = session.CreateConsumer(dest);
            if (method != null)
            {
                consumer.Listener += new MessageListener(OnObjectMessage);
            }
        }
        /// <summary>
        /// 消费事件
        /// </summary>
        /// <param name="receivedMsg"></param>
        private void OnTextMessage(IMessage receivedMsg)
        {
            if (receivedMsg != null)
            {
                Method.Invoke(Entity, new string[] { (receivedMsg as ITextMessage).Text });
            }
            else
            {
                Method.Invoke(Entity, null);
            }
        }
        /// <summary>
        /// 消费事件
        /// </summary>
        /// <param name="receivedMsg"></param>
        private void OnObjectMessage(IMessage receivedMsg)
        {
            if (receivedMsg != null)
            {
                Method.Invoke(Entity, new object[] { (receivedMsg as IObjectMessage).Body });
            }
            else
            {
                Method.Invoke(Entity, null);
            }
        }
        /// <summary>
        /// 注册消费者对象和事件
        /// </summary>
        /// <param name="entity"></param>
        /// <param name="method"></param>
        public void OnMessage(TEntity entity, MethodInfo method)
        {
            this.Entity = entity;
            this.Method = method;
        }
    }
}
